package zhongche;

import com.alibaba.fastjson.JSONObject;
import com.zhongche.bean.TrainInfo;
import com.zhongche.util.DateUtil;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;
import org.drools.core.io.impl.ClassPathResource;
import org.drools.template.model.Rule;
import org.kie.api.KieBase;
import org.kie.api.runtime.StatelessKieSession;
import org.kie.internal.utils.KieHelper;

import java.util.Map;

public class DemoClass {

    private static final String SPEED_KEY = "s0915691";


    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

//        env.fromCollection()

        String jsonStr = "{\n" +
                "  \"box_num\": 1,\n" +
                "  \"current_station\": 5,\n" +
                "  \"dDevice\": 50,\n" +
                "  \"dataParam\": {\n" +
                "\"s0000108\": 0,\n" +
                "\"s0915691\": 75,\n" +
                "    \"s1999915\": 1\n" +
                "  },\n" +
                "  \"data_sDeviceNo\": 20,\n" +
                "  \"frame_num\": 5657,\n" +
                "  \"line_id\": 2,\n" +
                "  \"line_num\": \"2\",\n" +
                "  \"message_type\": 5025,\n" +
                "  \"netLink_flag\": 0,\n" +
                "  \"next_station\": 0,\n" +
                "  \"ocs_time\": \"210716153627\",\n" +
                "  \"sDevice\": 40,\n" +
                "  \"sDeviceTime\": \"210716153626\",   \n" +
                "  \"second_circle\": 1,\n" +
                "  \"speed\": 0.0,\n" +
                "  \"start_station\": 5,\n" +
                "  \"terminal_station\": 28,\n" +
                "  \"train_id\": 159160,\n" +
                "  \"train_name\": \"159160\",\n" +
                "  \"train_num\": 159160,\n" +
                "  \"train_type\": \"A52\",\n" +
                "  \"ttype_id\": 10\n" +
                "}\n";
        String jsonStr2 = "{\n" +
                "  \"box_num\": 1,\n" +
                "  \"current_station\": 5,\n" +
                "  \"dDevice\": 50,\n" +
                "  \"dataParam\": {\n" +
                "\"s0000108\": 0,\n" +
                "\"s0915691\": 0,\n" +
                " \"s1999915\": 1\n" +
                "  },\n" +
                "  \"data_sDeviceNo\": 20,\n" +
                "  \"frame_num\": 5657,\n" +
                "  \"line_id\": 2,\n" +
                "  \"line_num\": \"2\",\n" +
                "  \"message_type\": 5025,\n" +
                "  \"netLink_flag\": 0,\n" +
                "  \"next_station\": 0,\n" +
                "  \"ocs_time\": \"210716153627\",\n" +
                "  \"sDevice\": 40,\n" +
                "  \"sDeviceTime\": \"210716155040\",   \n" +
                "  \"second_circle\": 1,\n" +
                "  \"speed\": 0.0,\n" +
                "  \"start_station\": 5,\n" +
                "  \"terminal_station\": 28,\n" +
                "  \"train_id\": 159160,\n" +
                "  \"train_name\": \"159160\",\n" +
                "  \"train_num\": 159160,\n" +
                "  \"train_type\": \"A52\",\n" +
                "  \"ttype_id\": 10\n" +
                "}\n";

        String jsonStr3 = "{\n" +
                "  \"box_num\": 1,\n" +
                "  \"current_station\": 5,\n" +
                "  \"dDevice\": 50,\n" +
                "  \"dataParam\": {\n" +
                "\"s0000108\": 0,\n" +
                "\"s0915691\": 60,\n" +
                "    \"s1999915\": 1\n" +
                "  },\n" +
                "  \"data_sDeviceNo\": 20,\n" +
                "  \"frame_num\": 5657,\n" +
                "  \"line_id\": 2,\n" +
                "  \"line_num\": \"2\",\n" +
                "  \"message_type\": 5025,\n" +
                "  \"netLink_flag\": 0,\n" +
                "  \"next_station\": 0,\n" +
                "  \"ocs_time\": \"210716153627\",\n" +
                "  \"sDevice\": 40,\n" +
                "  \"sDeviceTime\": \"210716173626\",   \n" +
                "  \"second_circle\": 1,\n" +
                "  \"speed\": 0.0,\n" +
                "  \"start_station\": 5,\n" +
                "  \"terminal_station\": 28,\n" +
                "  \"train_id\": 159160,\n" +
                "  \"train_name\": \"159160\",\n" +
                "  \"train_num\": 159160,\n" +
                "  \"train_type\": \"A52\",\n" +
                "  \"ttype_id\": 10\n" +
                "}\n";

        env.fromElements(jsonStr,jsonStr2,jsonStr3)
                .map(new MapFunction<String, TrainInfo>() {
                    @Override
                    public TrainInfo map(String value) throws Exception {
                        return JSONObject.parseObject(value, TrainInfo.class);
                    }
                })
                .name("转换对象")
                .keyBy(TrainInfo::getTrainId)
                .flatMap(new MyFloatMap())
                .print();

        env.execute();

//        env.addSource(new FlinkKafkaConsumer<Object>())

    }


    public static class MyFloatMap extends RichFlatMapFunction<TrainInfo, String> {


        private ValueState<Long> timeSate;




        @Override
        public void open(Configuration parameters) throws Exception {
            timeSate = getRuntimeContext().getState(new ValueStateDescriptor<Long>("time-state", Long.class));
        }

        @Override
        public void close() throws Exception {
            timeSate.clear();
        }

        @Override
        public void flatMap(TrainInfo value, Collector<String> out) throws Exception {
//            KieHelper kieHelper = new KieHelper();
//            kieHelper.kfs.write( new ClassPathResource( "rules/rules.drl") );
//            KieBase kbase = kieHelper.build();
//            StatelessKieSession statelessKieSession = kbase.newStatelessKieSession();
//            statelessKieSession.execute(value);

            Long timeSecond = timeSate.value();
            Long aLong = DateUtil.transDateStrToStamp(value.getsDeviceTime());
            if (aLong != null) {
                Map<String, Integer> dataParam = value.getDataParam();
                if (dataParam.get(SPEED_KEY) != null && dataParam.get(SPEED_KEY) == 0) {
                    timeSate.clear();
                    timeSate.update(aLong);
                } else {
                    if (timeSecond != null && dataParam.get(SPEED_KEY) > 0) {
                        Long intervalTime = (aLong - timeSecond) / 1000;
                        String message = value.getTrainName() + "(" + value.getTrainId() + "): 停车时间为" + intervalTime + "s";
                        out.collect(message);
                    }

                }
            }

        }




    }
}
